package com.bleeth.flow.core.util;

import cn.hutool.core.map.MapUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.spark.sql.SparkSession;

import java.net.URL;
import java.util.Iterator;
import java.util.Map;

/**
 * @author ：Bleeth
 * @date ：2021-08-05
 * @description：Spark工具类
 */
public class SparkUtil {

    private static SparkSession spark;
    private static SparkSession.Builder builder;
    private static Configuration config;

    static {
        builder = SparkSession
                .builder()
                .master("local[*]")
                .appName("Spark Flow Application...")
                .config("spark.ui.enabled", false);
    }

    /**
     * 配置spark环境
     *
     * @param
     * @return
     */
    public static void setConfig(String key, String value) {
        if (config == null) {
            config = new Configuration();
        }
        config.set(key, value);
    }

    /**
     * 配置spark环境
     * @param configMap
     */
    public static void setConfig(Map<String, String> configMap) {
        if (config == null) {
            config = new Configuration();
        }
        if (MapUtil.isNotEmpty(configMap)) {
            configMap.entrySet().stream().forEach(entry -> {
                String key = entry.getKey();
                String value = entry.getValue();
                config.set(key, value);
            });
        }
    }

    /**
     * 添加资源，比如hive-site.xml
     * @param url
     */
    public static void addResource(URL url) {
        if (config == null) {
            config = new Configuration();
        }
        config.addResource(url);
    }

    /**
     * 支持hive
     */
    public static void enableHiveSupport( ) {
       builder.enableHiveSupport();
    }


    /**
     * 获取spark环境
     *
     * @return
     */
    public static SparkSession getSparkInstance() {
        if (spark == null) {
            if (config != null) {
                Iterator<Map.Entry<String, String>> iterator = config.iterator();
                while (iterator.hasNext()){
                    Map.Entry<String, String> configEntry = iterator.next();
                    builder.config(configEntry.getKey(), configEntry.getValue());
                }
            }
            spark = builder.getOrCreate();
        }
        return spark;
    }

    /**
     * 销毁spark
     */
    public static void close() {
        spark.close();
    }

}
